-- 集群节点服务管理，注册和发现
-- auth:diandian
-- date:2018/9/27
--
--

local skynet = require "skynet.manager"
local cluster = require "skynet.cluster"
local ex_log = require "ex_log"

local _nodemap = {}
local _localmap = {}
local _center_host
local _followmap = {}
local CMD = {}

local function reload_cluster()
    local map = {}
    for k, v in pairs(_nodemap) do
        map[k] = v.host
    end
        cluster.reload(map)
end

function CMD.on_register_node(node)
    local exist = _nodemap[node.name]
    if exist and exist.host == node.host then
        return
    end
    _nodemap[node.name] = node
    reload_cluster()
end

function CMD.on_delete_node(name)
    _nodemap[name] = nil
    reload_cluster()
end

local function time_heartbeat()
    skynet.timeout(500, time_heartbeat)
    for k, v in pairs(_localmap) do
        cluster.send("center_node", ".ex_cluster_center",
            "heartbeat", v)
    end
end

function CMD.start(type_, name, host, center_host, follows)
    follows = follows or {}
    local followmap = {}
    for i, v in ipairs(follows) do
        followmap[v] = 1
        _followmap[v] = 1
    end
    if not _center_host then
        _center_host = center_host
        local t = {["center_node"] = center_host,}
        cluster.reload(t)
    end

    local err, node, nodemap = cluster.call("center_node", ".ex_cluster_center",
        "register_node", type_, name, host, followmap, _followmap)
    assert(not err, err)
    _nodemap = nodemap
    _nodemap[node.name] = node
    _localmap[node.name] = node
    reload_cluster()
    ex_log.info("cluster open ", node.name, node.host)
    cluster.open(node.name)
    skynet.timeout(500, time_heartbeat)
    return node.name
end

function CMD.get_info()
    return _nodemap
end

skynet.start(function()
    skynet.register(".ex_cluster_node")
    skynet.dispatch("lua", function(session, source, cmd, ...)
        local f = CMD[cmd]
        assert(f, cmd)
        skynet.retpack(f(...))
    end)
end)

